package com.hong.qrpc.handler.in.request;

import com.hong.qrpc.constant.ProtoTypeConst;
import com.hong.qrpc.constant.QrpcConst;
import com.hong.qrpc.factory.QrpcThreadFactory;
import com.hong.qrpc.factory.ResolverFactory;
import com.hong.qrpc.header.QrpcHeader;
import com.hong.qrpc.resolver.Resolver;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.SocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 处理请求的handler
 *
 * @Author HQT
 * @Date 2022/8/9 9:04
 */
@Slf4j
public class QrpcRequestHandler extends ChannelInboundHandlerAdapter {
    //经测试，ByteBuf的read系列方法的线程安全的，但是若有多次读取的话仍然线程不安全，需要进行同步处理
    private ByteBuf byteBuf = ByteBufAllocator.DEFAULT.directBuffer();
    private ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() << 1, Runtime.getRuntime().availableProcessors() << 1, 2L,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000), new QrpcThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());
    /**
     * 服务器端传来的请求处理对象
     */
    private com.hong.qrpc.RequestHandler handler;
    private SocketChannel socketChannel;
    private QrpcHeader header;
    private boolean isReadHeader = false;

    public QrpcRequestHandler(com.hong.qrpc.RequestHandler handler, SocketChannel socketChannel) {
        this.handler = handler;
        this.socketChannel = socketChannel;
    }

    /**
     * TODO 这是原本的、同步的处理方法，现在改为了使用线程池去处理请求以期通过将读入请求数据与处理请求数据的步骤分开的方式去提高响应速度
     * 改了之后发现效率略有提升，也没有发现线程不安全的问题，有待观察
     */
//    @Override
//    public void channelRead(ChannelHandlerContext ctx, Object msg) {
//        reentrantLock.lock();
//        Queue<Runnable> requests=new LinkedList<>();
//        ByteBuf tempByteBuf = null;
//        try {
//            Resolver resolver = ResolverFactory.getResolver(ProtoTypeConst.QRPC);
//            ByteBuf in = (ByteBuf) msg;
//            this.byteBuf.writeBytes(in);
//            in.release();
//
//            //可能会响应多个请求,将多个请求解析出来然后使用线程池进行异步响应
//            while (isFullRequest(this.byteBuf)) {
//                QrpcHeader header = (QrpcHeader) resolver.readHeader(this.byteBuf);
//                log.debug("got request header{}", header);
//                byte[] tempBytes = new byte[header.getDataLen()];
//                byteBuf.readBytes(tempBytes);
//                InputStream data = new ByteArrayInputStream(tempBytes);
//                ByteArrayOutputStream response = new ByteArrayOutputStream();
//
//                handler.onRequest(data, response, header.getSerializationType());
//
//                tempByteBuf = ByteBufAllocator.DEFAULT.buffer(QrpcConst.HEADER_BYTE_LEN + response.size());
//                QrpcHeader qrpcHeader = new QrpcHeader().setVersion(QrpcConst.QRPC_VERSION)
//                        .setDataLen(response.size()).setRequestId(header.getRequestId()).setSerializationType(header.getSerializationType());
//                log.debug("response header:{}", qrpcHeader);
//                tempByteBuf.writeBytes(resolver.toByteArr(qrpcHeader));
//                tempByteBuf.writeBytes(response.toByteArray());
//                socketChannel.writeAndFlush(tempByteBuf);
//            }
//        } finally {
//            if(tempByteBuf!=null){
//                tempByteBuf.release();
//            }
//            reentrantLock.unlock();
//        }
//    }

    /**
     * 对请求进行逐一解析，解析好的字节流传入服务端创建的请求对象中去获取请求结果，然后将请求结果写出
     *
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        Resolver resolver = ResolverFactory.getResolver(ProtoTypeConst.QRPC);
        ByteBuf in = (ByteBuf) msg;
        this.byteBuf.writeBytes(in);
        in.release();
        //可能会响应多个请求,将多个请求解析出来然后使用线程池进行异步响应
        while (isFullRequest(this.byteBuf)) {
            log.debug("got request header{}", this.header);
            byte[] tempBytes = new byte[this.header.getDataLen()];
            this.byteBuf.readBytes(tempBytes);
            InputStream data = new ByteArrayInputStream(tempBytes);
            QrpcHeader copyHeader = new QrpcHeader(this.header.getVersion(), this.header.getSerializationType()
                    , this.header.getRequestId(), this.header.getDataLen());
            poolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    ByteArrayOutputStream response = new ByteArrayOutputStream();

                    handler.onRequest(data, response, copyHeader.getSerializationType());
                    ByteBuf tempByteBuf = ByteBufAllocator.DEFAULT.buffer(QrpcConst.HEADER_BYTE_LEN + response.size());
                    QrpcHeader qrpcHeader = new QrpcHeader().setVersion(QrpcConst.QRPC_VERSION)
                            .setDataLen(response.size()).setRequestId(copyHeader.getRequestId()).setSerializationType(header.getSerializationType());
                    log.debug("response header:{}", qrpcHeader);
                    tempByteBuf.writeBytes(resolver.toByteArr(qrpcHeader));
                    tempByteBuf.writeBytes(response.toByteArray());
                    socketChannel.writeAndFlush(tempByteBuf);
                }
            });
        }
    }

    /**
     * @param src       被填充字段的byte数组
     * @param attribute 字段数组
     * @param startIdx  从哪个位置开始填充
     * @return
     */
    public int fillAttribute(byte[] src, byte[] attribute, int startIdx) {
        for (int i = 0; i < attribute.length; i++, startIdx++) {
            src[startIdx] = attribute[i];
        }
        return startIdx;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }

    /**
     * 使用此方法的方法是线程安全的，所以该方法也线程安全
     *
     * @param byteBuf
     * @return
     */
    private boolean isFullRequest(ByteBuf byteBuf) {
        Resolver resolver = ResolverFactory.getResolver(ProtoTypeConst.QRPC);
        if (isReadHeader) {
            //已经读了包头了，数据包没有读完整的情况
            if (byteBuf.readableBytes() >= header.getDataLen()) {
                //因为这个包在下次调用之前就读完了，下次就从头开始读起
                isReadHeader = false;
                return true;
            } else {
                return false;
            }
        } else {
            if (!(byteBuf.readableBytes() >= QrpcConst.HEADER_BYTE_LEN)) {
                return false;
            }
            header = (QrpcHeader) resolver.readHeader(this.byteBuf);
            isReadHeader = true;
            if (byteBuf.readableBytes() >= header.getDataLen()) {
                //因为这个包在下次调用之前就读完了，下次就从头开始读起
                isReadHeader = false;
                return true;
            } else {
                return false;
            }
        }
    }
}
